package com.lost.octopus.es.processor.mq.consumer.capture;

import com.lost.octopus.es.processor.mq.consumer.dto.ChangeDataCaptureInfo;
import com.lost.octopus.es.processor.mq.consumer.dto.SinkDeal;
import com.sunsharing.share.common.collection.MapUtil;
import lombok.extern.log4j.Log4j2;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.event.ApplicationStartedEvent;
import org.springframework.context.event.EventListener;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;

import static com.lost.octopus.es.processor.constant.enums.OperationType.DELETE;

/**
 * 抽象数据捕获处理到es实现
 * 按表的维度消费数据变更，若存在子配置，则使用parentId 进行in处理关联的子索引
 * 项目捕获处理清单见：<a href="https://alidocs.dingtalk.com/i/nodes/93NwLYZXWygx77mahXkY1X4bJkyEqBQm?iframeQuery=sheet_range%3Dkgqie6hm_5_1_1_1">es消费监控表清单</a>
 * @author zhangbh
 * @date 2025/4/23 10:37
 */
@Log4j2
@Component
public class DataCaptureManager extends BurstCaptureManager {

    @Resource
    private JdbcTemplate jdbcTemplate;
    @Resource
    private NamedParameterJdbcTemplate namedParameterJdbcTemplate;
    @Resource
    private DataCaptureIgnoreManager dataCaptureIgnoreManager;
    @Resource
    private DataCaptureConfigManager dataCaptureConfigManager;

    // 捕获策略
    private static final Map<String, Set<SinkDeal>> CAPTURE_STAGE = new HashMap<>();
    /**
     * 删除处理者策略
     */
    private static final Map<String, DataCaptureDeal> DEL_CAPTURE = new HashMap<>();

    @Autowired(required = false)
    List<DataCaptureDeal> deals;
    static boolean load = false;


    @EventListener(ApplicationStartedEvent.class)
    public void init() throws Exception {
        if (load) {
            return;
        }
        dataCaptureConfigManager.loadConfig(CAPTURE_STAGE);
        load = true;
    }

    private String getKey(ChangeDataCaptureInfo data) {
        return data.getDatabase() + "." + data.getTableName();
    }

    public boolean acceptCapture(ChangeDataCaptureInfo data) {
        // 数据处理,当前应该处理 xxx 表数据变更，查询出主表id，或者esId的sql
        String key = getKey(data);
        // 处理删除的情况
        if (acceptDelete(data, key)) {
            return true;
        }
        // 大部分都是更新
        Set<SinkDeal> configs = CAPTURE_STAGE.get(key);
        if (CollectionUtils.isEmpty(configs)) {
            log.warn("未找到表【{}】的变更捕获配置，请检查配置是否正确！！！", data.getTableName());
            return false;
        }
        // 处理配置规则
        sinkDeal(data, configs);
        return true;
    }

    private void sinkDeal(ChangeDataCaptureInfo data, Set<SinkDeal> configs) {
        for (SinkDeal info : configs) {
            // 初始化信息，并且获取到变更的sql
            String sql = info.getSql();
            List<String> ids;
            try {
                ids = jdbcTemplate.queryForList(sql, String.class, data.getPrimaryValue());
            } catch (Exception e) {
                log.error(String.format("捕获sql执行异常 desc %s sql %s primaryKey %s 请检查sql是否正确", info.getDesc(), sql, data.getPrimaryValue()), e);
                throw e;
            }
            log.info("开始处理捕获规则 sinkDeal {} sql:{},待处理数据:{},变更字段 {}", info.getDesc(), sql, ids.size(), data.getChangeFields());
            if (CollectionUtils.isEmpty(ids)) {
                log.warn(String.format("请确认关联sql注册正确，对应的表%s变更后，需要查询到id，用于检查新增/删除es数据！！！！！！！！！！！！", data.getTableName()));
            }
            // 分片处理,防止数据过大
            burstDataCapture(ids, info.getIdType(), info);
            // 处理子配置
            boolean canDealChildren = !CollectionUtils.isEmpty(info.getChildren()) && CollectionUtils.isNotEmpty(ids);
            // 检查忽略字段
            boolean ignoreField = dataCaptureIgnoreManager.ignoreChildSink(data, info);
            if (canDealChildren && !ignoreField) {
                sinkDealChildren(info, ids);
            }
        }
    }

    private void sinkDealChildren(SinkDeal parentInfo, List<String> ids) {
        Set<SinkDeal> children = parentInfo.getChildren();
        // 子处理，通过父id，进行 in 处理
        for (SinkDeal info : children) {
            String sql = info.getSql();
            if (CollectionUtils.isEmpty(ids)) {
                log.info("无需处理捕获规则 sinkDealChildren {} 使用parent 【{}】 sql:{},待处理数据:{} parentId is empty ", info.getDesc(), parentInfo.getDesc(), sql, 0);
                return;
            }
            // 获取子id
            List<String> childIds = queryChildrenIds(ids, info, parentInfo, sql);

            if (CollectionUtils.isEmpty(childIds)) {
                log.warn("请确认关联sql注册正确，desc : {} parentIds {}", info.getDesc(), childIds);
            }
            // 分片处理,防止数据过大
            burstDataCapture(childIds, info.getIdType(), info);
            // 处理子配置
            boolean canDealChildren = !CollectionUtils.isEmpty(info.getChildren()) && CollectionUtils.isNotEmpty(childIds);
            if (canDealChildren) {
                sinkDealChildren(info, childIds);
            }
        }
    }

    private List<String> queryChildrenIds(List<String> ids, SinkDeal info, SinkDeal parentInfo, String sql) {
        // 搜索出参数名
        String paramName = info.getInParamName();
        if (StringUtils.isBlank(sql)) {
            log.info("开始处理捕获规则 sinkDealChildren {} 使用parent 【{}】 ids,待处理数据:{}", info.getDesc(), parentInfo.getDesc(), ids.size());
            return ids;
        }
        List<String> childIds;
        try {
            childIds = namedParameterJdbcTemplate.queryForList(sql, MapUtil.ofHashMap(paramName, ids), String.class);
        } catch (Exception e) {
            log.error(String.format("捕获sql执行异常 children desc %s sql %s ids %s 请检查sql是否正确", info.getDesc(), sql, ids), e);
            throw e;
        }
        log.debug("开始处理捕获规则 sinkDealChildren {} parentIds={}", info.getDesc(), ids);
        log.info("开始处理捕获规则 sinkDealChildren {} 使用parent 【{}】 ids in sql:{},待处理数据:{}", info.getDesc(), parentInfo.getDesc(), sql, childIds.size());
        return childIds;
    }


    private boolean acceptDelete(ChangeDataCaptureInfo data, String key) {
        if (DELETE.eq(data.getType())) {
            DataCaptureDeal captureDeal = DEL_CAPTURE.get(key);
            if (captureDeal != null) {
                return captureDeal.acceptDelete(data);
            }
            return true;
        }
        return false;
    }


}
